package com.ksyun.campus.dataserver.services;

import com.ksyun.campus.common.constant.Sys;
import com.ksyun.campus.common.constant.ZkConstant;
import com.ksyun.campus.common.constant.api.DataServerApi;
import com.ksyun.campus.common.domain.DataServer;
import com.ksyun.campus.common.domain.ReplicaData;
import com.ksyun.campus.common.domain.StatInfo;
import com.ksyun.campus.common.util.ByteConvertUtil;
import com.ksyun.campus.dataserver.bootstrap.DataServerInstance;
import com.ksyun.campus.dataserver.util.FSHttpUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.tomcat.util.http.fileupload.IOUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.servlet.ServletInputStream;
import java.io.*;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Service
@Slf4j
public class DataService {

    @Autowired
    private CuratorFramework curatorFramework;

    private ConcurrentHashMap<String, FileOutputStream> outputMap = new ConcurrentHashMap<>();

    /**
     * 检查文件是否存在
     */
    public boolean check(String path) {
        path = path.replace(ZkConstant.META_DATA_PATH + "/", "");
        String name = path.replaceAll("/", "-");
        String realPath = DataServerInstance.getDataPath() + Sys.SEPARATOR + name;
        File file = new File(realPath);
        if (file.exists()) {
            return true;
        }
        return false;
    }

    /**
     * 删除文件
     *
     * @param fileSystem
     * @param path
     * @return
     */
    public boolean delete(String fileSystem, String path, int size) {
        log.info("删除文件~");
        String name = (fileSystem + path).replaceAll("/", "-");
        String realPath = DataServerInstance.getDataPath() + Sys.SEPARATOR + name;
        File file = new File(realPath);
        if (file.exists()) {
            DataServerInstance.updateInfo(-1, -size);
            //如果存在则删除
            file.delete();
            return true;
        }
        return false;
    }


    /**
     * 关闭流
     *
     * @param fileSystem
     * @param path
     */
    public List<ReplicaData> close(String fileSystem, String path) {
        String name = (fileSystem + path).replaceAll("/", "-");
        FileOutputStream outputStream = outputMap.get(name);
        if (outputStream != null) {
            try {
                DataServerInstance.updateInfo(1, 0);
                outputStream.close();
                outputMap.remove(name);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        List<ReplicaData> replicaDataList = getReplicaNode(path);
        List<ReplicaData> newList = new ArrayList<>(replicaDataList);
        //保留副本
        syncWrite(replicaDataList, fileSystem, path);
        addLocalNode(newList, path);
        return newList;
    }


    @Async
    public void syncWrite(List<ReplicaData> replicaDataList, String fileSystem, String path) {
        FileInputStream read = read(fileSystem, path);
        if (read == null) {
            return;
        }
        byte[] bytes = null;
        try {
            bytes = new byte[read.available()];
            read.read(bytes);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                read.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        for (ReplicaData replicaData : replicaDataList) {
            syncWriteQuery(replicaData.dsNode, bytes, fileSystem, path);
        }
    }

    public void syncWriteQuery(String addres, byte[] data, String fileSystem, String path) {
        log.info("同步数据~{}", addres);
        FSHttpUtils.write(addres + DataServerApi.SYNC_DATA, data, path, fileSystem, 0, data.length);
    }

    /**
     * 写
     *
     * @param in
     * @param fileSystem
     * @param path
     * @param offset
     * @param length
     * @return
     */
    public void write(InputStream in, String fileSystem, String path, int offset, int length) {
        //todo 写本地
        String name = (fileSystem + path).replaceAll("/", "-");
        FileOutputStream out = outputMap.get(name);
        if (out != null) {
            try {
                IOUtils.copy(in, out);
            } catch (IOException e) {
                e.printStackTrace();
            }
            DataServerInstance.updateInfo(0, length);
            return;
        }
        String realPath = DataServerInstance.getDataPath() + Sys.SEPARATOR + name;
        File file = new File(realPath);
        try {
            out = new FileOutputStream(file);
            outputMap.put(name, out);
            IOUtils.copy(in, out);
            DataServerInstance.updateInfo(0, length);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
            //todo
        } catch (IOException e) {
            e.printStackTrace();
            //todo
        }
    }


    /**
     * 读
     *
     * @param fileSystem
     * @param path
     * @return
     */
    public FileInputStream read(String fileSystem, String path) {
        String name = (fileSystem + path).replaceAll("/", "-");
        String realPath = DataServerInstance.getDataPath() + Sys.SEPARATOR + name;
        File file = new File(realPath);
        FileInputStream inputStream = null;
        try {
            inputStream = new FileInputStream(file);
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
        return inputStream;
    }

    /**
     * 获取副本节点
     *
     * @param path
     * @return
     */
    private List<ReplicaData> getReplicaNode(String path) {
        List<ReplicaData> list = new ArrayList<>();
        List<DataServer> dataList = getNodeDataList();
        for (DataServer dataServer : dataList) {
            ReplicaData replicaData = new ReplicaData();
            replicaData.dsNode = dataServer.getIp() + ":" + dataServer.getPort();
            replicaData.id = dataServer.getId();
            replicaData.path = path;
            list.add(replicaData);
        }
        return list;
    }

    private void addLocalNode(List<ReplicaData> list, String path) {
        DataServer dataServer = DataServerInstance.getDataServer();
        ReplicaData replicaData = new ReplicaData();
        replicaData.dsNode = dataServer.getIp() + ":" + dataServer.getPort();
        replicaData.path = path;
        replicaData.id = dataServer.getId();
        list.add(replicaData);
    }

    /**
     * 数据同步写入
     *
     * @param data
     * @param fileSystem
     * @param path
     * @param offset
     * @param length
     */
    public void sync(ServletInputStream data, String fileSystem, String path, int offset, int length) {
        log.info("副本同步~");
        String name = (fileSystem + path).replaceAll("/", "-");
        String realPath = DataServerInstance.getDataPath() + Sys.SEPARATOR + name;
        File file = new File(realPath);
        FileOutputStream out = null;
        FileLock lock = null;
        try {
            out = new FileOutputStream(file);
            FileChannel channel = out.getChannel();
            lock = channel.lock();
            IOUtils.copy(data, out);
            DataServerInstance.updateInfo(1, length);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
            //todo
        } catch (IOException e) {
            e.printStackTrace();
            //todo
        } finally {
            try {
                if (lock != null) {
                    lock.release();
                }
                out.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * 数据修复
     *
     * @param node  需要修复的节点
     * @param cpath 完整的路径 /metadata/volume/{普通显示的路径}  -> 需要变成  volume-aaa-bbb.jpg 截去 /metadata/
     */
    public void repair(List<String> node, String cpath) {
        cpath = cpath.replace(ZkConstant.META_DATA_PATH + "/", "");
        String[] split = cpath.split("/", 2);
        String fileSystem = split[0];
        String path = split[1];
        FileInputStream read = read(fileSystem, "/" + path);
        if (read == null) {
            return;
        }
        byte[] bytes = null;
        try {
            bytes = new byte[read.available()];
            read.read(bytes);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                read.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        for (String ad : node) {
            syncWriteQuery(ad, bytes, fileSystem, "/" + path);
        }
    }


    /**
     * 获取数据节点
     *
     * @return
     */
    private List<DataServer> getNodeDataList() {
        List<String> dataNodeList = getDataNodeList();
        List<DataServer> dataServers = new ArrayList<>();
        DataServer dataServer = DataServerInstance.getDataServer();
        String localNode = dataServer.getIp() + ":" + dataServer.getPort();
        for (String addre : dataNodeList) {
            //排除本节点
            if (!addre.equals(localNode)) {
                dataServers.add(getDataServerInZk(ZkConstant.DATA_SERVER + "/" + addre));
            }
        }
        //不够剩下的两份，先返回
        if (dataServers.size() < 2) {
            return dataServers;
        }
        List<DataServer> collect = dataServers.stream().sorted((a, b) -> {
            return a.getUseCapacity() - b.getUseCapacity();
        }).limit(2).collect(Collectors.toList());
        return collect;
    }

    /**
     * 获取所有的数据服务节点
     *
     * @return
     */
    private List<String> getDataNodeList() {
        try {
            if (this.curatorFramework.checkExists().forPath(ZkConstant.DATA_SERVER) != null) {
                return this.curatorFramework.getChildren().forPath(ZkConstant.DATA_SERVER);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return new ArrayList<>();
    }

    private DataServer getDataServerInZk(String nodePath) {
        byte[] bytes = new byte[0];
        try {
            bytes = curatorFramework.getData().forPath(nodePath);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return (DataServer) ByteConvertUtil.toObj(bytes, DataServer.class);
    }


}
